package com.ervanalb.cellradio.shared.communicator;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;

import com.ervanalb.cellradio.client.Informational;
import com.ervanalb.cellradio.shared.communicator.packets.AckPacketData;
import com.ervanalb.cellradio.shared.communicator.packets.Packet;
import com.ervanalb.cellradio.shared.communicator.packets.PacketData;

/**
 * A fully self-contained object for handling communication between a server and
 * client in either direction. Adds a handshaking layer to the underlying
 * PacketSource and PacketSink which may be unreliable.
 * 
 * Instantiating a Communicator starts a new thread, which listens for incoming
 * messages. When it receives them, it notifies the default listener if the
 * packet was send asynchronously, or the relevant ResponseListeners if the
 * packet is a reply. Handshaking is done automatically.
 * 
 * Additionally, provides the method safeSend which can be used for sending a
 * packet with handshaking. This is done synchronously, and blocks until
 * handshaking is received or times out, so it may be slow.
 * 
 * @author eric
 */
public class Communicator extends Thread {

	public static final long ACK_TIMEOUT = 100;
	public static final long RX_TICK = 100;
	public static final int NUM_TRIES = 3;

	public int listenPort;
	public int speakPort;

	public Exception e;
	protected boolean go;

	protected HashMap<Integer, List<PacketListener>> replyListeners;

	protected PacketListener defaultListener;

	public PacketSource source;
	public PacketSink sink;

	protected Object txInProgress;

	public int txNum, rxNum;

	public static final int DGRAM_BUF_SIZE = 1024;

	public Communicator(PacketSource source, PacketSink sink,
			PacketListener defaultListener) {
		e = null;

		this.source = source;
		this.sink = sink;

		this.defaultListener = defaultListener;

		replyListeners = new HashMap<Integer, List<PacketListener>>();

		txInProgress = new Object();

		reset();

		start();
	}

	public void reset() {
		synchronized (txInProgress) {
			rxNum = 1;
			txNum = 1;
		}
	}

	public void run() {
		Packet receivedPacket;

		go = true;

		while (go) {
			try {
				receivedPacket = source.receive(RX_TICK);
				if (receivedPacket == null)
					continue;

				checkPacket(receivedPacket);

			} catch (IOException e) {
				this.e = e;
				break;
			}
		}
	}

	public void finish() {
		go = false;
		interrupt();
	}

	public void validPacket(Packet p) {
		List<PacketListener> listOfListeners;

		synchronized (replyListeners) {
			if (p.replyTo == 0) {
				defaultListener.processPacket(p);
			} else if (replyListeners.containsKey(p.replyTo)) {
				listOfListeners = replyListeners.get(p.replyTo);
				for (PacketListener l : listOfListeners)
					l.processPacket(p);
			} else if (p.pd.type == PacketData.TYPE_ERROR) {
				defaultListener.processPacket(p); // Process error replies even
													// if there is no handler
			} else {
				System.out.println("Warning: response dropped on floor: " + p);
			}
		}
	}

	public void checkPacket(Packet p) throws IOException {
		if (p.pd.type == PacketData.TYPE_ACK) {
			validPacket(p);
		} else if (p.pd.type == PacketData.TYPE_CONNECT
				|| p.pd.type == PacketData.TYPE_CONNECTED) {
			// These packets don't abide by sequence numbers, so validate them
			// always
			sendAck(p);
			validPacket(p);
		} else if (p.number == rxNum) {
			sendAck(p);
			rxNum++;
			validPacket(p);
		} else if (p.number < rxNum) {
			sendAck(p); // Duplicate packet
		} else {
			sendAck(p, (byte) 0);
		}
	}

	public void addReplyListener(ReplyListener rl) {
		synchronized (replyListeners) {
			if (replyListeners.containsKey(rl.message)) {
				replyListeners.get(rl.message).add(rl);
			} else {
				List<PacketListener> newList = new LinkedList<PacketListener>();
				newList.add(rl);
				replyListeners.put(rl.message, newList);
			}
		}
	}

	public void delReplyListener(ReplyListener rl) {
		synchronized (replyListeners) {
			if (replyListeners.containsKey(rl.message)) {
				List<PacketListener> list = replyListeners.get(rl.message);
				if (list.contains(rl)) {
					list.remove(rl);
					if (list.size() == 0)
						replyListeners.remove(rl.message);
				}
			}
		}
	}

	// Packet does not have txnumber set
	// Returns: completely null result if expectReply==false and it succeeds
	// ReplyListener wrapped in an informational if expectReply==true and it
	// succeeds
	// ReplyListener string if it fails
	public Informational<ResponseListener> safeSend(Packet p,
			boolean expectReply) throws IOException {
		synchronized (txInProgress) {
			AckListener ackListener;
			ResponseListener responseListener = null;

			Informational<ResponseListener> i = null;
			// NULL indicates success, which is a poor choice for the default
			// value. However, i is guaranteed to be modified

			int tries;

			p.number = txNum;
			ackListener = new AckListener(this, txNum);
			if (expectReply) {
				responseListener = new ResponseListener(this, txNum);
			}
			for (tries = 0; tries < NUM_TRIES; tries++) {
				sink.send(p);
				Packet response = ackListener.getReply(ACK_TIMEOUT);
				if (response != null) {
					if (response.pd.parseAck().success == 1) {
						// We have succeeded
						txNum++;
						if (expectReply) {
							// A responselistener indicates success when we are
							// expecting a reply
							return new Informational<ResponseListener>(
									responseListener);
						} else {
							// Null indicates success when we aren't expecting a
							// reply
							return null;
						}
					}
					i = new Informational<ResponseListener>("Got NAK");
					break;
				}
				System.out.println("Attempt " + (tries + 1)
						+ " failed to get ACK for " + p);
			}

			// If we get here, we have failed
			if (tries == NUM_TRIES) {
				i = new Informational<ResponseListener>("Tried " + tries
						+ " times, never got ACK, giving up.");
			}
			if (expectReply)
				responseListener.suicide();
			return i;
		}

	}

	public void sendAck(Packet p) throws IOException {
		sendAck(p, (byte) 1);
	}

	public void sendAck(Packet p, byte success) throws IOException {
		if (p.pd.type == PacketData.TYPE_ACK) // Do not ACK ACKs
			return;
		Packet ack = new Packet(0, p.channel, p.number, new AckPacketData(
				success).pack(), p.addr);
		sink.send(ack);
	}
}
